package com.moli.iov.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;

public abstract class KafkaConsumerServer implements AcknowledgingMessageListener<String,Object> {

    protected final Logger logger=LoggerFactory.getLogger("KafkaConsumerServer");

    @Override
    public void onMessage(ConsumerRecord<String, Object> record, Acknowledgment acknowledgment) {
       // logger.info("=============kafkaConsumer开始手动提交消费=============");
        String topic = record.topic();
        String key = record.key();
        Object value =record.value();
        long offset = record.offset();
        int partition = record.partition();
        manualHandle(record,acknowledgment);
       // logger.info("topic:"+topic+" value:"+value+" key:"+key+" offset:"+offset+" partition:"+partition);
       // logger.info("~~~~~~~~~~~~~kafkaConsumer手动消费结束~~~~~~~~~~~~~");
    }

    /**
     *
     * @param record
     * @param acknowledgment
     */

    public abstract void manualHandle(ConsumerRecord<String, Object> record,Acknowledgment acknowledgment);

}
